package main

import (
	"fmt"
	"gopkg.in/ini.v1"
	"logagent/elastic"
	"logagent/getlog"
	"logagent/kafka"
	"logagent/redis"
	"os"
	"strings"
	"time"
)

var (
	topic string
)

func main() {
	//加载配置文件
	cfg, err := ini.Load("./config/config.ini")
	if err != nil {
		fmt.Printf("Fail to read file: %v", err)
		os.Exit(1)
	}

	//典型读取操作，默认分区可以使用空字符串表示
	ipaddr := cfg.Section("kafka").Key("ipaddr").String()
	topic = cfg.Section("kafka").Key("topic").String()
	esurl := cfg.Section("elastic").Key("url").String()
	filelist := strings.Replace(cfg.Section("logfile").Key("filelist").String(), " ", "", -1)

	FileList := strings.Split(filelist, ",")
	//FileList := []string{"D:/learndj/test.log", "D:/learndj/event.log", "D:/learndj/package.log"}

	//1.初始化kafka链接

	err = kafka.InitProducer([]string{ipaddr})
	if err != nil {
		fmt.Printf("Init Kafka failed, err:%v", err)
		return
	}

	elastic.InitElastic(esurl)

	kafka.InitConsumer([]string{ipaddr}, topic, "test01")

	redisHost := cfg.Section("redis").Key("ipaddr").String()
	redisPass := cfg.Section("redis").Key("password").String()
	LocalBool, _ := cfg.Section("logfile").Key("collectlocal").Bool()

	if LocalBool {
		//读取配置文件中的日志文件
		for _, filename := range FileList {
			tailobj, err := getlog.TailfInit(filename)
			if err != nil {
				fmt.Printf("get tailobj failed,err: %s", err)
			}
			go getlog.ReadLine(tailobj, filename, topic)
		}
	} else {
		key := cfg.Section("logfile").Key("localip").String()
		//watch redis key【filename】的值 如果发生变化 就更新
		redis.InitClient(redisHost, redisPass)
		err = redis.Rdb.Set(key, "/var/log/messages", time.Hour*1000).Err()
		if err != nil {
			panic(err)
		}
		var (
			QueueListOld []string
			QueueListNew []string
		)

		QueueListOld = redis.WatchFunct(key)
		getlog.Ergodic(topic, QueueListOld)
		go func() {
			for {
				QueueListNew = redis.WatchFunct(key)
				//fmt.Println("本次的目录有：", QueueListNew)

				if strings.Join(QueueListNew, "") != strings.Join(QueueListOld, "") {
					QueueListOld = QueueListNew
					getlog.Ergodic(topic, QueueListNew)
				}

			}
		}()
	}

	for {
		select {
		case msg, ok := <-kafka.Consumer.Messages():
			if ok {
				hostname := cfg.Section("logfile").Key("hostname").String()
				ipaddr := cfg.Section("logfile").Key("localip").String()
				//fmt.Printf(" kafka msg: %s \n", msg.Value)
				elastic.IndexElastic(hostname, ipaddr, string(msg.Value))
			}
		}
	}

}
